package com.atguigu.flink;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.client.program.StreamContextEnvironment;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.table.descriptors.Schema;
import org.apache.flink.table.expressions.Expression;
import org.apache.flink.table.types.DataType;
import org.apache.flink.util.Collector;

import java.util.HashMap;
import java.util.Iterator;


public class TestWC {
    public static void main(String[] args) throws Exception {
        //1.创建一个流的执行环境(get方法可以根据实际所处的环境来执行各种模式）
        StreamExecutionEnvironment env = StreamContextEnvironment.getExecutionEnvironment();
        //设置分区数
        env.setParallelism(1);

        //2.通过执行环节得到一个流
        DataStreamSource<String> source = env.readTextFile("input/words.txt");

        //3.对流进行各种转换
        SingleOutputStreamOperator<String> source1 = source.flatMap(new FlatMapFunction<String, String>() {
            @Override
            public void flatMap(String line, Collector<String> collector) throws Exception {
                String[] split = line.split(" ");
                for (String s1 : split) {
                    collector.collect(s1);
                }
            }
        });

        SingleOutputStreamOperator<Tuple2<String, Integer>> operator = source1.map(new MapFunction<String, Tuple2<String, Integer>>() {
            @Override
            public Tuple2<String, Integer> map(String value) throws Exception {
                return Tuple2.of(value, 1);
            }
        });

        KeyedStream<Tuple2<String, Integer>, String> keyedStream = operator.keyBy(new KeySelector<Tuple2<String, Integer>, String>() {
            @Override
            public String getKey(Tuple2<String, Integer> value) throws Exception {
                return value.f0;
            }
        });

        //按照元组的第二个数据进行累加（sum）
        SingleOutputStreamOperator<Tuple2<String, Integer>> sum = keyedStream.sum(1);
        HashMap<String, Schema> map = new HashMap<>();

        //4.输出流数据
        sum.print();

        SingleOutputStreamOperator<String> process = source.keyBy(value -> JSON.parseObject(value).getString("table"))
                .process(new ProcessFunction<String, String>() {
                    @Override
                    public void open(Configuration parameters) throws Exception {
                        super.open(parameters);
                    }

                    @Override
                    public void processElement(String value, Context ctx, Collector<String> out) throws Exception {
                        Schema schema;
                        JSONObject files = JSON.parseObject(value).getJSONObject("files");
                        schema = new Schema();
                        Iterator<String> iterator = files.keySet().iterator();
                        while (iterator.hasNext()) {
                            String key = iterator.next();
                            String values = files.getString(key);
                            schema.field(key, back(values));
                            map.put(files.getString("table"), schema);
                        }

                    }
                });

        Schema schema = new Schema();
        StreamTableEnvironment.create(env).registerDataStream("44",source,"a,x,v");
        //5.执行执行环境
        env.execute();
    }

    private static DataType back(String va){
        if (va == "string"){
            return DataTypes.STRING();
        }
        return null;
    }
}
